package com.vanke.agent.sync;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.log4j.Logger;
import com.vanke.agent.ADataSource;
import com.vanke.agent.C;
import com.vanke.agent.Cfg;
import com.vanke.agent.common.CstUtil;
import com.vanke.agent.common.Util;
import com.vanke.agent.common.log.LoggerUtil;

public class OpportunitySync extends AgentSync{
	Logger log = LoggerUtil.getLoggerByName("opportunity_sync");
	protected Set<String> aready = new HashSet<String>();
	private Map<String, Long> cache=new HashMap<String, Long>();
	@Override
	public void sync(){
		log.info("=============sync new Opportunity===========");
		Connection con = ADataSource.getConnection(ADataSource.SCHEMA_MY);
		Connection wcon = ADataSource.getConnection(ADataSource.SCHEMA_BIZ);
		long now = System.currentTimeMillis();
		String table = "s_Opportunity";
		String fields = C.FIELDS_OPPORTUNITY;
		
		String timeStr = dateFormat.format(Util.offMinute(opportunityCfg.beginTime,
				rebackOffset));
		String nowStr = dateFormat.format(new Date(System.currentTimeMillis()));
		PreparedStatement pstmt =null;
		PreparedStatement wpstmt=null;
		PreparedStatement dataUpdatePs=null;
		ResultSet rs = null;
		try {
			log.info("Begin check GUID mapping");
			loadGUIDMapping();
			log.info("End check GUID mapping");
//			 缓存现有的OppGUID
			loadGUID(wcon);
			String querySql = "SELECT opp.*,oc.cstGUID from "+table+" opp join s_Opp2Cst oc on opp.OppGUID=oc.oppGUID "
							 +" where "
							 +" (opp.ModifyOn is null and opp.CreatedOn >= ? AND opp.CreatedOn < ? ) "
							 +" OR "
							 +" (opp.ModifyOn >= ? AND opp.ModifyOn < ? ) "
							 +" order by CASE WHEN ModifyOn Is NULL Then CreatedOn Else ModifyOn End ";
			log.info("execute: " + querySql + "[" + timeStr + "~" + nowStr
					+ "]");
		    pstmt = con.prepareStatement(querySql,
					ResultSet.TYPE_SCROLL_INSENSITIVE,
					ResultSet.CONCUR_READ_ONLY);
			
			pstmt.setString(1, timeStr);
			pstmt.setString(2, nowStr);
			pstmt.setString(3, timeStr);
			pstmt.setString(4, nowStr);
			 rs = pstmt.executeQuery();
			int count = 0;
			int update = 0;
			if (!rs.next()) {
				log.info("there are no data in table Opportunity!!!");
				ADataSource.close(rs);
				ADataSource.close(pstmt);
				return;
			}
//			rs.last(); //移到最后一行  
//			int rowCount = rs.getRow(); //得到当前行号，也就是记录数  
//			log.info("sync record size : " + rowCount);
//			rs.first(); //如果还要用结果集，就把指针再移到初始化的位置  
//			wcon.setAutoCommit(false);
			String insertSql = CstUtil
					.formatInsertSql(table, fields+",oriGUID,cstGUID");
			 wpstmt = wcon.prepareStatement(insertSql);
			log.info("execute insert : " + insertSql);
			int columns = fields.split(",").length;
			String updateSql = CstUtil
					.formatUpdateSql(table, fields+",oriGUID,cstGUID");
			updateSql += " where OppGUID =? and oriGUID=? ";
			log.info("execute update : " + updateSql);
			 dataUpdatePs = wcon.prepareStatement(updateSql);
			
			do {
				Timestamp modiDate = rs.getTimestamp("ModifyOn");
				opportunityCfg.beginTime = new java.sql.Date(modiDate!=null?modiDate.getTime():rs.getTimestamp("CreatedOn").getTime());
				for (int i = 1; i <= columns; i++) {
					wpstmt.setString(i, rs.getString(i));
					dataUpdatePs.setString(i, rs.getString(i));
				}
//
//				try {
					String oppGUID=rs.getString("OppGUID").trim();
					String oriCstGUID=rs.getString("cstGUID");
					long uid = cache.get(oriCstGUID)==null?0:cache.get(oriCstGUID);
//					long uid = CstUtil.getCstGUIDByOriGuid(wcon, oriCstGUID);
//					if (uid == 0) {
//						// 查不到，则到明源查义主档记录的身份证或手机，再反查数据中心
//						log.info("Opportunity no CstGUID found for:" + oriCstGUID
//								+ ", retrieve again");
//						uid = CstUtil.retrieveMemship(oriCstGUID);
//					}
					dataUpdatePs.setString(columns+1,oriCstGUID);
					dataUpdatePs.setLong(columns+2,uid);
					wpstmt.setString(columns+1,oriCstGUID);
					wpstmt.setLong(columns+2,uid);
					if (aready.contains(oppGUID+"_"+oriCstGUID)) {
						dataUpdatePs.setString(columns+3, rs.getString("OppGUID"));
						dataUpdatePs.setString(columns+4, oriCstGUID);
						dataUpdatePs.executeUpdate();
//						dataUpdatePs.addBatch();
						update++;
					} else {
						wpstmt.executeUpdate();
//						wpstmt.addBatch();
						count++;
					}
					if ((count + update) % 2000 == 0) {
						log.info(count + " added and " + update + " updated, spend "
							+ (System.currentTimeMillis() - now) + "ms");
					}
/*					// 无论batchSize是多少，最多一次提交不能超过10000
					if ((count + update) % 2000 == 0) {
						wpstmt.executeBatch();
						log.info(1);
						dataUpdatePs.executeBatch();
						log.info(2);
						wcon.commit();
						log.info(3);
//						 保存配置，最后一次的时间
						Cfg.set(contractCfg.cfgKey,
								dateFormat.format(contractCfg.beginTime));
						wpstmt.clearBatch();
						dataUpdatePs.clearBatch();

						log.info(count + " added and " + update + " updated, spend "
								+ (System.currentTimeMillis() - now) + "ms");
					} // end if
*///				} catch (Exception e) {
//					e.printStackTrace();
//					log.error(e);
//				}

			} while (rs.next());
			/*wpstmt.executeBatch();
			log.info(1);
			dataUpdatePs.executeBatch();
			log.info(2);
			wcon.commit();
			log.info(3);
//			 保存配置，最后一次的时间
			Cfg.set(contractCfg.cfgKey,
					dateFormat.format(contractCfg.beginTime));
			wpstmt.clearBatch();
			dataUpdatePs.clearBatch();*/
			
//			 保存配置，最后一次的时间
			Cfg.set(opportunityCfg.cfgKey,
					dateFormat.format(opportunityCfg.beginTime));
			Cfg.save();
			now = System.currentTimeMillis() - now;
			log.info(table + " all " + count + " added and " + update
					+ " updated, spend " + now + "ms");
			// 关闭
//			ADataSource.close(rs);
//			ADataSource.close(pstmt);
//			ADataSource.close(wpstmt);
//			ADataSource.close(dataUpdatePs);

//			log.info(update + " Opportunity has updated ");
//			log.info(count + " Opportunity has saved ");
		

		} catch (Exception e) {
			e.printStackTrace();
			log.error("Opportunity Sync Error："+e.getMessage(),e);
		} finally {
			ADataSource.close(rs,pstmt,con);
			ADataSource.close(dataUpdatePs);
			ADataSource.close(null,wpstmt,wcon);
			aready.clear();
			cache.clear();
		}
		log.info("=============sync Opportunity end===========");
	} // ~sync
	/* 缓存已有的GUID */
	private void loadGUID(Connection con) throws SQLException {
		Statement stmt = null;
		ResultSet rs = null;
		try {
			stmt = con.createStatement();
			rs = stmt.executeQuery("select concat(oppGUID,'_',oriGUID) from s_Opportunity");
			while (rs.next()) {
				aready.add(rs.getString(1));
			}
		} catch (SQLException e) {
			e.printStackTrace();
			throw e;
		} finally {
			ADataSource.close(rs, stmt, null);
		}
	} //~s
	/**
	 * 加载已有的GUID映射
	 */
	 void loadGUIDMapping() throws SQLException {
		List<String> mappingTables = retrieveMappingTables();
		log.info("guid mapping tables: " + mappingTables);
		Connection con = ADataSource.getConnection(ADataSource.SCHEMA_BIZ);
		try {
			Statement stmt = con.createStatement();
			long count = 0;
			for (String table : mappingTables) {
				String sql = "select OriGUID,CstGUID from " + table;
				ResultSet rs = stmt.executeQuery(sql);
				while (rs.next()) {
					count++;
					String guid = rs.getString(1);
					long uid = rs.getLong(2);
					cache.put(guid, uid);
				}
				ADataSource.close(rs);
				log.info(count + " guid-uid mapping cached");
			}
			ADataSource.close(stmt);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			ADataSource.close(con);
		}
	}
		// 加载所有guid的映射表
		static List<String> retrieveMappingTables() throws SQLException {
			List<String> tables = new ArrayList<String>(10);
			Connection con = ADataSource.getConnection(ADataSource.SCHEMA_BIZ);

			Statement stmt = con.createStatement();

			ResultSet rs = stmt.executeQuery("show tables");
			while (rs.next()) {
				String table = rs.getString(1);
				if (table.toLowerCase().startsWith("cst_cstguidmapping_")) {
					tables.add(table);
				}
			}
			ADataSource.close(rs, stmt, con);
			return tables;
		}
}
